package com.example.sink;

import com.example.beans.SensorReading;
import com.example.source.SensorSource;
import com.example.source.SourceTest4_UDF;
import org.apache.flink.api.common.serialization.Encoder;
import org.apache.flink.api.common.serialization.SimpleStringEncoder;
import org.apache.flink.core.fs.Path;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.sink.filesystem.StreamingFileSink;
import org.apache.flink.streaming.api.functions.sink.filesystem.rollingpolicies.DefaultRollingPolicy;
import org.apache.flink.streaming.api.operators.StreamSink;

import java.util.concurrent.TimeUnit;


/**
 *
 *  * @projectName myflinkstu
 *  * @title     Test0_FileSink
 *  * @package    com.example.sink
 *  * @description
 *  * @author hjx
 *  * @date   2022-3-22 10:02
 *  * @version V1.0.0
 *  * @copyright 2022 ty
 *
 */
public class Test0_FileSink {

    public static void main(String[] args) {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(4);

        DataStreamSource<SensorReading> inputStream = env.addSource(new SensorSource());

        StreamingFileSink<String> fileSink = StreamingFileSink
                .<String>forRowFormat(new Path("./fileSinkOutput"),
                        new SimpleStringEncoder<>("UTF-8"))
                .withRollingPolicy(
                        DefaultRollingPolicy.builder()
                                // 设置滚动策略, 至少包含 15 分钟的数据 时 会开启新的文件，将之前的内容归档保存, 滚动分区文件
                                .withRolloverInterval(TimeUnit.MINUTES.toMillis(15)
                                )
                                 // 最近 5 分钟没有收到新的数据
                                .withInactivityInterval(TimeUnit.MINUTES.toMillis(5
                                ))
                                // 文件大小已达到 1 GB
                                .withMaxPartSize(1024 * 1024 * 1024)
                                .build())
                .build();
        // 将 Event 转换成 String 写入文件
        inputStream.map(SensorReading::toString).addSink(fileSink);
        try {
            env.execute();
        } catch (Exception e) {
            e.printStackTrace();
        }


    }

}
